{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "3e4dd44e",
   "metadata": {},
   "outputs": [],
   "source": [
    "# demo\n",
    "# 导入row类\n",
    "from pyspark.sql import Row\n",
    "# 导入数据类型\n",
    "from pyspark.sql.types import *\n",
    "# 导入SparkSession\n",
    "from pyspark.sql import SparkSession"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "963dcdbf",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "zhangsan\n",
      "lisi\n",
      "wangwu\n",
      "zhaoliu\n"
     ]
    }
   ],
   "source": [
    "# 使用Row类型生成一行数据\n",
    "r1 = Row(1,'zhangsan',20)\n",
    "r2 = Row(2,'lisi',20)\n",
    "\n",
    "#从行中可以通过下标取值\n",
    "print(r1[1])\n",
    "print(r2[1])\n",
    "\n",
    "\n",
    "# 使用Row类型生成一行数据可以指定字段名\n",
    "r3 = Row(id=3,name='wangwu',age=20)\n",
    "r4 = Row(id=4,name='zhaoliu',age=20)\n",
    "\n",
    "#从行中可以通过字段名取值\n",
    "print(r3['name'])\n",
    "print(r4['name'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "9a31f4e5",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 定义schema信息\n",
    "# 使用StructType类进行定义\n",
    "# add()方法是指定字段信息\n",
    "# 第一参数，字段名\n",
    "# 第二个参数，字段信息\n",
    "# 第三个参数是否允许为空值  默认是True，允许为空\n",
    "schema_type = StructType().\\\n",
    "    add('id',IntegerType()).\\\n",
    "    add('name',StringType()).\\\n",
    "    add('age',IntegerType(),False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "178406ff",
   "metadata": {},
   "outputs": [],
   "source": [
    "# dataframe的创建方法是由SparkSession提供的,需要生成SparkSession对象\n",
    "# 是一个固定写法\n",
    "spark = SparkSession.builder.getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "910766f9",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+----+---+\n",
      "| id|name|age|\n",
      "+---+----+---+\n",
      "|  1|张三| 20|\n",
      "|  2|李四| 30|\n",
      "+---+----+---+\n",
      "\n",
      "root\n",
      " |-- id: integer (nullable = true)\n",
      " |-- name: string (nullable = true)\n",
      " |-- age: integer (nullable = false)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# 定义每行数据\n",
    "r1= Row(id=1,name='张三',age=20)\n",
    "r2= Row(id=2,name='李四',age=30)\n",
    "\n",
    "# 定义schema信息\n",
    "schema_type = StructType().\\\n",
    "    add('id',IntegerType()).\\\n",
    "    add('name',StringType()).\\\n",
    "    add('age',IntegerType(),False)\n",
    "\n",
    "# 创建dataframe数据\n",
    "# 使用sparksession对象中的createDataFrame方法创建\n",
    "# 第一个参数，指定行数据，将每行数据放入列表中\n",
    "# 第二参数，指定表信息\n",
    "# 创建成功后就可以得到一个df类型的数据\n",
    "df = spark.createDataFrame([r1,r2],schema_type)\n",
    "\n",
    "\n",
    "# 查看df的表数据\n",
    "# 使用dataframe下的show方法\n",
    "df.show()\n",
    "\n",
    "# 查看schema信息\n",
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "d2539fb1",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+--------+---+\n",
      "| id|    name|age|\n",
      "+---+--------+---+\n",
      "|  1|zhangsan| 20|\n",
      "|  2|    lisi| 20|\n",
      "+---+--------+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# SparkSession中可以通过sparkContex获取sparkContex对象\n",
    "sc = spark.sparkContext\n",
    "\n",
    "# 生成rdd\n",
    "rdd = sc.parallelize([[1,'zhangsan',20],[2,'lisi',20]])\n",
    "\n",
    "# 定义schema信息\n",
    "schema_type = StructType().\\\n",
    "    add('id',IntegerType()).\\\n",
    "    add('name',StringType()).\\\n",
    "    add('age',IntegerType(),False)\n",
    "\n",
    "# toDF 将二维rdd数据转为dataframe数据\n",
    "df = rdd.toDF(schema_type)\n",
    "\n",
    "# 查看df数据\n",
    "df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "b0d7e2fe",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[Row(id=1, name='zhangsan', age=20), Row(id=2, name='lisi', age=20)]\n",
      "['zhangsan', 'lisi']\n"
     ]
    }
   ],
   "source": [
    "new_rdd = df.rdd\n",
    "print(new_rdd.collect())\n",
    "map_rdd = new_rdd.map(lambda x: x['name'])\n",
    "print(map_rdd.collect())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3aab44d8",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
